package com.fir.proto.mqtt;

import com.fir.proto.config.MqttConfig;
import com.fir.proto.protobuf.info.MesInfo;
import com.google.protobuf.Descriptors;
import com.google.protobuf.Message;
import com.hivemq.client.mqtt.datatypes.MqttQos;
import com.hivemq.client.mqtt.mqtt3.Mqtt3BlockingClient;
import com.hivemq.client.mqtt.mqtt3.Mqtt3Client;
import com.hivemq.client.mqtt.mqtt3.message.publish.Mqtt3Publish;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Lazy;

import javax.annotation.Resource;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.function.Consumer;


/**
 * 使用mqtt tcp形式的连接方式
 * @author fir
 */
@Configuration
@Slf4j
public class MqttClient {


    @Resource
    private MqttConfig mqttConfig;



    /**
     * mqtt
     */
    private Mqtt3BlockingClient mqttBean;

    @Autowired
    @Qualifier("mqttBean")
    public void setCoalStackPositionInfoService(@Lazy Mqtt3BlockingClient mqttBean) {
        this.mqttBean = mqttBean;
    }


    /**
     * 测试数据
     */
    public final String TEST_MSG = "/test";


    /**
     * MQTT 服务
     *
     * @param mqttBack 接收到的 MQTT 消息的回调函数
     * @return MQTT 阻塞式（Blocking）客户端
     */
    @Bean(name = "mqttBean")
    public Mqtt3BlockingClient mqttBean(@Qualifier("mqttBackBean") Consumer<Mqtt3Publish> mqttBack) {
        log.info("[MQTT] 开始连接");
        Mqtt3BlockingClient client = Mqtt3Client.builder()
                .identifier(UUID.randomUUID().toString())
                .serverHost(mqttConfig.getIp())
                .serverPort(mqttConfig.getPort())
                .addConnectedListener(context -> log.info("[MQTT] 连接成功 {}", context))
                .addDisconnectedListener(context -> log.info("[MQTT] 断开连接 {}", context))
                .automaticReconnectWithDefaultConfig()
                .simpleAuth()
                .username(mqttConfig.getUsername())
                .password(mqttConfig.getPassword().getBytes())
                .applySimpleAuth()
                .buildBlocking();

        if (mqttConfig.isConnectMqtt()) {
            client.connect();
            log.info("[MQTT] 连接成功");
        }

        // 数据订阅
        client.toAsync().subscribeWith()
                .topicFilter(TEST_MSG)
                .qos(MqttQos.AT_LEAST_ONCE)
                .callback(mqttBack)
                .send();
        log.info("[MQTT] 注册成功");
        return client;
    }



    @Bean(name = "mqttBackBean")
    public Consumer<Mqtt3Publish> mqttBackBean() {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String datetime = sdf.format(new Date());
        log.info("[MQTT]:" + datetime);
        Consumer<Mqtt3Publish> callback = mqtt3Publish -> {
            String topic;
            topic = mqtt3Publish.getTopic().toString();

            byte[] mqttData = mqtt3Publish.getPayloadAsBytes();

            log.info("[MQTT] 主题{} 消息体 {}}", topic, Arrays.toString(mqttData));

            try {
                MesInfo mesInfo =  MesInfo.parseFrom(mqttData);
                System.out.println("mesInfo============================================================");
                System.out.println(mesInfo.toString());
                System.out.println(mesInfo);
                System.out.println("mesInfo============================================================");

                // 打印所有字段
                for (Map.Entry<Descriptors.FieldDescriptor, Object> entry : mesInfo.getAllFields().entrySet()) {
                    Descriptors.FieldDescriptor field = entry.getKey();
                    Object value = entry.getValue();
                    System.out.println(field.getName() + ": " + value);
                }
            } catch (Exception e) {
                e.printStackTrace();
                // pass
            }
        };
        log.info("[MQTT] 处理结束:" + datetime);
        return callback;
    }


    /**
     * mqtt 数据发送
     *
     * @param topic 订阅题目
     * @param msg   消息
     */
    public void send(String topic, byte[] msg) {
        mqttBean.publishWith()
                .topic(topic)
                .payload(msg)
                .qos(MqttQos.AT_LEAST_ONCE)
                .send();
        log.info("mqtt 发送 {} {}:", topic, msg);
    }

}


